package com.sfx.mq.demo;

import com.sfx.mq.constants.MQConstant;
import com.sfx.mq.constants.TopicProperties;
import com.sfx.mq.dao.po.Order;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.junit.Test;

import java.util.Arrays;
import java.util.Date;
import java.util.List;

/**
 * Created with IntelliJ IDEA.
 * Description: 发送批量消息
 * User: sfx
 * Date: 2023-10-27
 * Time: 15:39
 */
public class OrderlyMessageTest {
    // 搞一组订单
    public List<Order> getMessageList() {
        return Arrays.asList(
                new Order("jdkd", "高性能MySQL", 1, "下单"),
                new Order("jdkd", "高性能MySQL", 1, "物流"),
                new Order("jdkd", "高性能MySQL", 1, "签收"),

                new Order("tmkd", "深入理解java并发编程", 2, "下单"),
                new Order("tmkd", "深入理解java并发编程", 2, "物流"),
                new Order("tmkd", "深入理解java并发编程", 2, "签收")
        );
    }

    @Test
    public void testProducer() throws Exception {
        // 创建生产者
        DefaultMQProducer producer = new DefaultMQProducer("test-producer-group");
        // 指定nameSrv
        producer.setNamesrvAddr(MQConstant.NAME_SRV);
        producer.start();
        // 发送消息
        List<Order> messageList = getMessageList();
        messageList.forEach(order -> {
            Message message = new Message(TopicProperties.ORDERLY_TOPIC, order.toString().getBytes());
            try {
                producer.send(message, new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> list, Message message, Object orderNo) {
                        // 返回把消息放到哪一个队列,这样相同订单就放到了一个队列中
                        int index = orderNo.hashCode() % list.size();
                        return list.get(index);
                    }
                }, order.getOrderNo());
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
        System.out.println(new Date());
        System.out.println("发送了批量消息");
        producer.shutdown();
    }

    @Test
    public void testConsumer() throws Exception {
        // 创建消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-consumer-group");
        // 连接nameSrv
        consumer.setNamesrvAddr(MQConstant.NAME_SRV);
        // 订阅主题的消息,可以进行过滤
        consumer.subscribe(TopicProperties.ORDERLY_TOPIC, "*");
        // 创建一个消息监听器,监听队列中的消息
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                MessageExt messageExt = list.get(0);
                System.out.println(new String(messageExt.getBody()));
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        // 开启消费者
        consumer.start();
        // 挂起JVM,不要停
        System.in.read();
    }
}